package com.orderpay_detect;

import com.orderpay_detect.bean.OrderEvent;
import com.orderpay_detect.bean.OrderResult;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @Description: TODO QQ1667847363
 * @author: xiao kun tai
 * @date:2021/11/12 16:10
 *  订单支付实时监控
 *  使用 Process Function 实现
 */
public class OrderTimeoutWithoutCep {
    //定义超时事件的侧输出流标签
    private final static OutputTag<OrderResult> orderTimeoutTag = new OutputTag<OrderResult>("order-timeout") {
    };


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //从文件中读取数据
        String filePath = "OrderPayDetect/src/main/resources/OrderLog.csv";
        DataStream<String> fileStream = env.readTextFile(filePath);

        //将读取的数据转换为POJO类型
        DataStream<OrderEvent> orderEventStream = fileStream
                .map(line -> {
                    String[] fields = line.split(",");
                    return new OrderEvent(new Long(fields[0]), fields[1], fields[2], new Long(fields[3]));
                })
                .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<OrderEvent>() {
                    @Override
                    public long extractAscendingTimestamp(OrderEvent orderEvent) {
                        return orderEvent.getTimestamp() * 1000L;
                    }
                });

        //自定义处理函数，主流输出正常匹配订单事件，侧输出流输出超时报警事件
        SingleOutputStreamOperator<OrderResult> resultStream = orderEventStream
                .keyBy(OrderEvent::getOrderId)
                .process(new OrderPayMatchDetect());

        resultStream.print("payed normally");
        resultStream.getSideOutput(orderTimeoutTag).print("timeout");

        env.execute("order timeout detect without cep job");

    }

    /**
     * 自定义KeyedProcessFunction
     */
    public static class OrderPayMatchDetect extends KeyedProcessFunction<Long, OrderEvent, OrderResult> {
        //定义状态，保存之前订单是否已经create，pay的事件

        ValueState<Boolean> isPayedState;
        ValueState<Boolean> isCreatedState;

        //定义状态，保存定时器的时间戳
        ValueState<Long> timerTsState;

        @Override
        public void open(Configuration parameters) throws Exception {
            isPayedState = getRuntimeContext()
                    .getState(new ValueStateDescriptor<Boolean>("is-payed", Boolean.class, false));

            isCreatedState = getRuntimeContext()
                    .getState(new ValueStateDescriptor<Boolean>("is-created", Boolean.class, false));

            timerTsState = getRuntimeContext()
                    .getState(new ValueStateDescriptor<Long>("timer-ts", Long.class));
        }

        @Override
        public void processElement(OrderEvent orderEvent, Context context, Collector<OrderResult> collector) throws Exception {

            //先获取当前状态
            Boolean isPayed = isPayedState.value();
            Boolean isCreated = isCreatedState.value();
            Long timerTs = timerTsState.value();

            //判断当前事件类型
            if ("create".equals(orderEvent.getEventType())) {
                //如果当前是create，需判断是否支付过
                if (isPayed) {
                    //如果已经正常支付，输出正常匹配结果
                    collector.collect(new OrderResult(orderEvent.getOrderId(), "payed successfully"));
                    //清空状态，删除定时器
                    isCreatedState.clear();
                    isPayedState.clear();
                    timerTsState.clear();
                    context.timerService().deleteEventTimeTimer(timerTs);
                } else {
                    //如果没有支付过，注册15分钟后的定时器，开始等待支付事件
                    Long ts = (orderEvent.getTimestamp() + 15*60) * 1000L;
                    context.timerService().registerEventTimeTimer(ts);

                    //更新状态
                    timerTsState.update(ts);
                    isCreatedState.update(true);

                }
            } else if ("pay".equals(orderEvent.getEventType())) {
                //如果来的是pay，需判断是否有下单事件来过
                if (isCreated) {
                    //已经有下单的事件，需继续判断支付的时间戳是否超过国5分钟
                    if (orderEvent.getTimestamp() * 1000L < timerTs) {
                        //在15分钟内，没有超时，正常匹配
                        collector.collect(new OrderResult(orderEvent.getOrderId(), "payed successfully"));
                    } else {
                        //已经超时，侧输出流输出报警
                        context.output(orderTimeoutTag, new OrderResult(orderEvent.getOrderId(), "payed but already timeout"));
                    }
                    //清空状态，删除定时器
                    isCreatedState.clear();
                    isPayedState.clear();
                    timerTsState.clear();
                    context.timerService().deleteEventTimeTimer(timerTs);

                } else {
                    //没有下单事件，乱序，注册一个定时器，等待下单事件
                    context.timerService().registerEventTimeTimer(orderEvent.getTimestamp() * 1000L);
                    timerTsState.update(orderEvent.getTimestamp() * 1000L);
                    isPayedState.update(true);
                }
            }

        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<OrderResult> out) throws Exception {
            //定时器触发，说明一定有一个事件没来
            if (isPayedState.value()) {
                //如果pay来了，说明create没来
                ctx.output(orderTimeoutTag, new OrderResult(ctx.getCurrentKey(), "payed but not found created log"));
            } else {
                //如果pay没来，支付超时
                ctx.output(orderTimeoutTag, new OrderResult(ctx.getCurrentKey(), "timeout"));
            }
            //清空状态，删除定时器
            isCreatedState.clear();
            isPayedState.clear();
            timerTsState.clear();
        }
    }
}
